package com.demo.common.redis.stream;

import cn.hutool.extra.spring.SpringUtil;
import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * <p>
 * 在容器启动后处理 Stream 类型的消息订阅对象
 * </p>
 *
 * @author molong
 * @date 2022/6/22
 */
public class StreamApplicationStartedEvent implements ApplicationListener<ApplicationStartedEvent> {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Override
    public void onApplicationEvent(ApplicationStartedEvent event) {
        StreamMessageListenerContainer container = SpringUtil.getBean(StreamMessageListenerContainer.class);
        groupConsumer(container);
        normalConsumer(container);
        //启动监听容器
        container.start();
        //如果是广播消费，不设置分组
        //container.receive(StreamOffset.latest("mq"), new ExampleStreamListener3());
    }

    /**
     * 普通消费者
     * @param container 消费者监听容器
     */
    private void normalConsumer(StreamMessageListenerContainer container) {
        Map<String, NormalStreamListener> beansOfType = SpringUtil.getBeansOfType(NormalStreamListener.class);
        List<NormalStreamListener> values = beansOfType.values().stream().collect(Collectors.toList());
        for (NormalStreamListener value : values) {
            container.receive(StreamOffset.create(value.getFullStreamKey(), value.getReadOffset()), value);
        }
    }

    /**
     * 处理分组消费组
     *
     * @param container 消费者监听容器
     */
    private void groupConsumer(StreamMessageListenerContainer container) {
        //处理分组消费
        Map<String, GroupStreamListener> beansOfType = SpringUtil.getBeansOfType(GroupStreamListener.class);
        List<GroupStreamListener> values = beansOfType.values().stream().collect(Collectors.toList());
        Map<String, List<String>> mqGroups = new HashMap<>(values.size());
        for (GroupStreamListener value : values) {
            if (!contains(value.getFullStreamKey(), mqGroups, value.getConsumer().getGroup())) {
                redisTemplate.opsForStream().createGroup(value.getFullStreamKey(), value.getConsumer().getGroup());
            }
            //构建一个流读取请求， 不自动确认，并且在发送错误时任务继续执行
            StreamMessageListenerContainer.StreamReadRequest streamReadRequest = StreamMessageListenerContainer.StreamReadRequest
                    .builder(StreamOffset.create(value.getFullStreamKey(), value.getReadOffset()))
                    .consumer(value.getConsumer())
                    .autoAcknowledge(false)
                    .cancelOnError(t -> false)
                    .build();
            //container.receive(value.getConsumer(), StreamOffset.create(value.getFullStreamKey(), value.getReadOffset()), value);
            container.register(streamReadRequest, value);
        }
    }

    /**
     * 是否存在对应的分组，不存在则创建
     *
     * @param streamKey key
     * @param mqGroups  分组
     * @param group     组名
     * @return 是否存在
     */
    private boolean contains(String streamKey, Map<String, List<String>> mqGroups, String group) {
        if (!mqGroups.containsKey(streamKey) && redisTemplate.hasKey(streamKey)) {
            StreamInfo.XInfoGroups infoGroups = redisTemplate.opsForStream().groups(streamKey);
            List<String> groups = infoGroups.stream().map(StreamInfo.XInfoGroup::groupName).collect(Collectors.toList());
            mqGroups.put(streamKey, groups);
        }
        if (CollectionUtils.isEmpty(mqGroups.get(streamKey))) {
            return false;
        }
        return mqGroups.get(streamKey).contains(group);
    }
}
